package com.google.code.stalkernet.node;

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Queue;
import java.util.Random;
import java.util.Set;
import java.util.logging.Logger;

import com.google.code.stalkernet.advertisement.Advertisement;
import com.google.code.stalkernet.advertisements.Advertisements;
import com.google.code.stalkernet.advertisements.event.AdvertisementsEvent;
import com.google.code.stalkernet.advertisements.event.AdvertisementsListener;
import com.google.code.stalkernet.database.Database;
import com.google.code.stalkernet.database.event.DatabaseEvent;
import com.google.code.stalkernet.database.event.DatabaseListener;
import com.google.code.stalkernet.entry.Entry;
import com.google.code.stalkernet.entry.EntryID;
import com.google.code.stalkernet.message.AdvertisementsAckMessage;
import com.google.code.stalkernet.message.AdvertisementsMessage;
import com.google.code.stalkernet.message.EntriesMessage;
import com.google.code.stalkernet.message.EntriesRequestMessage;
import com.google.code.stalkernet.message.LeafIDsAckMessage;
import com.google.code.stalkernet.message.LeafIDsMessage;
import com.google.code.stalkernet.message.MessageUtils;
import com.google.code.stalkernet.message.MessageVisitor;
import com.google.code.stalkernet.peer.Peer;
import com.google.code.stalkernet.peer.event.PeerEvent;
import com.google.code.stalkernet.peer.event.PeerListener;

/**
 * Simple implementation of {@link Node} interface.
 *
 * @author Mikhail Vladimirov
 */
public class SimpleNode extends AbstractNode
{
    private final static Logger logger =
        Logger.getLogger (SimpleNode.class.getName ());

    /**
     * Maximum number of requested, but not yet received entries per peer.
     */
    public final static int MAX_REQUESTED_ENTRIES = 256;

    private final NodeID id;

    private final Database database;
    private final Advertisements advertisements;

    private final Map <EntryID, Entry> hangingEntries = new HashMap <> ();
    private final Map <EntryID, Peer> requestedEntryIDs = new HashMap <> ();
    private final Map <EntryID, Collection <Entry>> missingParents =
        new HashMap <> ();
    private final Map <EntryID, Integer> missingParentsCounts =
        new HashMap <> ();

    private final Map <Peer, PeerWrapper> peers = new HashMap <> ();

    /**
     * Create new node that operates on top of given database.
     *
     * @param database database to operate on top of
     * @param advertisements advertisements to use
     */
    public SimpleNode (
        Database database,
        Advertisements advertisements)
    {
        if (database == null)
            throw new IllegalArgumentException ("Database is null");

        if (advertisements == null)
            throw new IllegalArgumentException (
                "Advertisements is null");

        byte [] idBytes = new byte [16];
        new Random ().nextBytes (idBytes);

        id = NodeUtils.createNodeID (idBytes);

        this.database = database;
        this.advertisements = advertisements;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public NodeID getID ()
    {
        return id;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void addPeer (Peer peer)
    {
        if (peer == null)
            throw new IllegalArgumentException ("Peer is null");

        if (peers.containsKey (peer))
            throw new IllegalStateException ("Peer already added: " + peer);

        if (id.equals (peer.getRemoteNodeID ()))
        {
            logger.warning ("Connected to myself: " + peer);
            peer.disconnect ();
            return;
        }

        for (Peer otherPeer: peers.keySet ())
        {
            if (otherPeer.getRemoteNodeID ().equals (peer.getRemoteNodeID ()))
            {
                logger.warning ("Already connected to " + peer);
                peer.disconnect ();
                return;
            }
        }

        PeerWrapper peerWrapper = new PeerWrapper (peer);

        peers.put (peer, peerWrapper);
        firePeerAdded (peer);

        peerWrapper.start ();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void removePeer (Peer peer)
    {
        if (peer == null)
            throw new IllegalArgumentException ("Peer is null");

        PeerWrapper peerWrapper = peers.get (peer);

        if (peerWrapper == null)
            throw new IllegalStateException ("Peer does not exists: " + peer);

        peerWrapper.stop ();

        peers.remove (peer);
        firePeerRemoved (peer);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Peer [] getAllPeers ()
    {
        return peers.keySet ().toArray (new Peer [peers.size ()]);
    }

    private void entriesReceived (Entry ... entries)
    {
        if (entries == null)
            throw new IllegalArgumentException ("Entries is null");

        Collection <Entry> addedHangingEntries = new ArrayList <> ();

        Queue <Entry> validEntries = new ArrayDeque <> ();

        for (Entry entry: entries)
        {
            EntryID entryID = entry.getID ();

            requestedEntryIDs.remove (entryID);

            if (!database.hasEntry (entryID) &&
                !hangingEntries.containsKey (entryID))
            {
                hangingEntries.put (entryID, entry);

                int missingParentsCount = 0;
                for (EntryID parentID: entry.getParentIDs ())
                {
                    if (!database.hasEntry (parentID))
                    {
                        missingParentsCount += 1;

                        Collection <Entry> children =
                            missingParents.get (parentID);

                        if (children == null)
                        {
                            children = new ArrayList <> ();
                            missingParents.put (parentID, children);
                        }

                        children.add (entry);
                    }
                }

                missingParentsCounts.put (
                    entryID, Integer.valueOf (missingParentsCount));

                if (missingParentsCount == 0)
                    validEntries.add (entry);
                else
                    addedHangingEntries.add (entry);
            }
        }

        for (Entry validEntry; (validEntry = validEntries.poll ()) != null;)
        {
            EntryID validEntryID = validEntry.getID ();

            hangingEntries.remove (validEntryID);
            missingParentsCounts.remove (validEntryID);
            Collection <Entry> children = missingParents.remove (validEntryID);
            if (children != null)
            {
                for (Entry child: children)
                {
                    EntryID childID = child.getID ();
                    int missingParentsCount =
                        missingParentsCounts.get (childID).intValue ();
                    missingParentsCount -= 1;
                    missingParentsCounts.put (
                        childID, Integer.valueOf (missingParentsCount));

                    if (missingParentsCount == 0)
                        validEntries.add (child);
                }
            }

            database.addEntry (validEntry);
        }

        for (PeerWrapper peerWrapper: peers.values ())
            peerWrapper.hangingEntriesAdded (
                addedHangingEntries.toArray (
                    new Entry [addedHangingEntries.size ()]));
    }

    private class PeerWrapper
        implements PeerListener, MessageVisitor <Void>, DatabaseListener,
        AdvertisementsListener
    {
        private final Peer peer;

        private final Set <EntryID> sentLeafIDs = new HashSet <> ();
        private final Set <Advertisement> sentAdvertisements =
            new HashSet <> ();

        private final Set <EntryID> knownEntryIDs = new HashSet <> ();
        private final Set <EntryID> requestableEntryIDs = new HashSet <> ();

        private int requestedEntries = 0;
        private boolean leafIDsSent = false;
        private boolean advertisementsSent = false;

        public PeerWrapper (Peer peer)
        {
            if (peer == null)
                throw new IllegalArgumentException ("Peer is null");

            this.peer = peer;
        }

        public void start ()
        {
            peer.addPeerListener (this);
            database.addDatabaseListener (this);
            advertisements.addAdvertisementsListener (this);

            sendLeafIDs ();
            sendAdvertisements ();
        }

        public void stop ()
        {
            peer.removePeerListener (this);
            database.removeDatabaseListener (this);
            advertisements.removeAdvertisementsListener (this);

            if (requestedEntries > 0)
            {
                Collection <EntryID> entryIDs =
                    new ArrayList<> (requestedEntries);

                for (Map.Entry <EntryID, Peer> entry:
                    requestedEntryIDs.entrySet ())
                {
                    if (peer.equals (entry.getValue ()))
                        entryIDs.add (entry.getKey ());
                }

                int count = entryIDs.size ();
                if (count > 0)
                {
                    EntryID [] entryIDsArray =
                        entryIDs.toArray (new EntryID [count]);

                    for (PeerWrapper peerWrapper: peers.values ())
                        if (peerWrapper != this)
                            peerWrapper.entriesUnrequested (entryIDsArray);
                }
            }
        }

        @Override // PeerListener
        public void messageReceived (PeerEvent peerEvent)
        {
            if (peerEvent == null)
                throw new IllegalArgumentException ("Peer event is null");

            peerEvent.getMessage ().accept (this);
        }

        @Override // PeerListener
        public void disconnected (PeerEvent peerEvent)
        {
            removePeer (peer);
        }

        @Override // MessageVisitor <Void>
        public Void visitLeafIDsMessage (LeafIDsMessage leafIDsMessage)
        {
            if (leafIDsMessage == null)
                throw new IllegalArgumentException ("Leaf IDs message is null");

            addKnownEntryIDs (leafIDsMessage.getLeafIDs ());

            peer.sendMessage (MessageUtils.createLeafIDsAckMessage ());

            return null;
        }

        @Override // MessageVisitor <Void>
        public Void visitLeafIDsAckMessage (LeafIDsAckMessage leafIDsAckMessage)
        {
            if (leafIDsAckMessage == null)
                throw new IllegalArgumentException (
                    "Leaf IDs ack message is null");

            leafIDsSent = false;

            sendLeafIDs ();

            return null;
        }

        @Override // MessageVisitor <Void>
        public Void visitEntriesRequestMessage (
            EntriesRequestMessage entriesRequestMessage)
        {
            if (entriesRequestMessage == null)
                throw new IllegalArgumentException (
                    "Entries request message is null");

            EntryID [] entryIDs = entriesRequestMessage.getEntryIDs ();
            int count = entryIDs.length;
            Entry [] entries = new Entry [count];

            for (int i = 0; i < count; i++)
                entries [i] = database.getEntry (entryIDs [i]);

            peer.sendMessage (MessageUtils.createEntriesMessage (entries));

            return null;
        }

        @Override // MessageVisitor <Void>
        public Void visitEntriesMessage (EntriesMessage entriesMessage)
        {
            if (entriesMessage == null)
                throw new IllegalArgumentException ("Entries message is null");

            Entry [] entries = entriesMessage.getEntries ();

            entriesReceived (entries);

            requestedEntries -= entries.length;

            sendEntriesRequest ();

            return null;
        }

        @Override // MessageVisitor <Void>
        public Void visitAdvertisementsMessage (
            AdvertisementsMessage advertisementsMessage)
        {
            if (advertisementsMessage == null)
                throw new IllegalArgumentException (
                    "Advertisement message is null");

            for (Advertisement advertisement:
                advertisementsMessage.getAdvertisements ())
                fireAdvertisementReceived (peer, advertisement);

            peer.sendMessage (
                MessageUtils.createAdvertisementsAckMessage ());

            return null;
        }

        @Override // MessageVisitor <Void>
        public Void visitAdvertisementsAckMessage (
            AdvertisementsAckMessage advertisementsAckMessage)
        {
            if (advertisementsAckMessage == null)
                throw new IllegalArgumentException (
                    "Advertisements ack message is null");

            advertisementsSent = false;

            sendAdvertisements ();

            return null;
        }

        @Override // DatabaseListener
        public void entryAdded (DatabaseEvent databaseEvent)
        {
            if (databaseEvent == null)
                throw new IllegalArgumentException ("Database event is null");

            Entry entry = databaseEvent.getEntry ();
            EntryID entryID = entry.getID ();

            if (knownEntryIDs.remove (entryID))
                requestableEntryIDs.remove (entryID);

            sendLeafIDs ();
        }

        @Override // AdvertisementsListener
        public void advertisementAdded (AdvertisementsEvent advertisementsEvent)
        {
            if (advertisementsEvent == null)
                throw new IllegalArgumentException (
                    "Adevertisements event is null");

            sendAdvertisements ();
        }

        @Override // AdvertisementsListener
        public void advertisementRemoved (
            AdvertisementsEvent advertisementsEvent)
        {
            // Do nothing
        }

        public void hangingEntriesAdded (Entry ... entries)
        {
            if (entries == null)
                throw new IllegalArgumentException ("Entries is null");

            Set <EntryID> knownParentIDs = new HashSet <> ();

            for (Entry entry: entries)
            {
                EntryID entryID = entry.getID ();

                if (knownEntryIDs.contains (entryID)) // If we known the entry
                {
                    requestableEntryIDs.remove (entryID);

                    // Then we know its parents!
                    knownParentIDs.addAll (
                        Arrays.asList (entry.getParentIDs ()));
                }
            }

            addKnownEntryIDs (knownParentIDs.toArray (
                new EntryID [knownParentIDs.size ()]));
        }

        public void entriesRequested (EntryID ... entryIDs)
        {
            if (entryIDs == null)
                throw new IllegalArgumentException ("Entry IDs is null");

            for (EntryID entryID: entryIDs)
                requestableEntryIDs.remove (entryID);
        }

        public void entriesUnrequested (EntryID ... entryIDs)
        {
            if (entryIDs == null)
                throw new IllegalArgumentException ("Entry IDs is null");

            for (EntryID entryID: entryIDs)
            {
                if (knownEntryIDs.contains (entryID) &&
                    !hangingEntries.containsKey (entryID))
                    requestableEntryIDs.add (entryID);
            }

            sendEntriesRequest ();
        }

        private void addKnownEntryIDs (EntryID ... entryIDs)
        {
            if (entryIDs == null)
                throw new IllegalArgumentException ("Entry IDs is null");

            for (EntryID entryID: entryIDs)
            {
                if (!database.hasEntry (entryID))
                {
                    if (knownEntryIDs.add (entryID))
                    {
                        if (!hangingEntries.containsKey (entryID) &&
                            !requestableEntryIDs.contains (entryID))
                            requestableEntryIDs.add (entryID);
                    }
                }
            }

            sendEntriesRequest ();
        }

        private void sendLeafIDs ()
        {
            if (!leafIDsSent)
            {
                Set <EntryID> leafIDs =
                    new HashSet <> (Arrays.asList (database.getLeafIDs ()));

                sentLeafIDs.retainAll (leafIDs);
                leafIDs.removeAll (sentLeafIDs);
                sentLeafIDs.addAll (leafIDs);

                if (!leafIDs.isEmpty ())
                {
                    peer.sendMessage (
                        MessageUtils.createLeafIDsMessage (
                            leafIDs.toArray (new EntryID [leafIDs.size ()])));

                    leafIDsSent = true;
                }
            }
        }

        private void sendAdvertisements ()
        {
            if (!advertisementsSent)
            {
                Set <Advertisement> advertisementsSet =
                    new HashSet <> (
                        Arrays.asList (
                            advertisements.getAllAdvertisements ()));

                sentAdvertisements.retainAll (advertisementsSet);
                advertisementsSet.removeAll (sentAdvertisements);
                sentAdvertisements.addAll (advertisementsSet);

                if (!advertisementsSet.isEmpty ())
                {
                    peer.sendMessage (
                        MessageUtils.createAdvertisementsMessage (
                            advertisementsSet.toArray (
                                new Advertisement [
                                    advertisementsSet.size ()])));

                    advertisementsSent = true;
                }
            }
        }

        private void sendEntriesRequest ()
        {
            int maxToRquest = MAX_REQUESTED_ENTRIES - requestedEntries;

            Set <EntryID> entryIDs = new HashSet <> ();

            Iterator <EntryID> i = requestableEntryIDs.iterator ();
            while (i.hasNext () && entryIDs.size () < maxToRquest)
                entryIDs.add (i.next ());

            int count = entryIDs.size ();

            if (count > 0)
            {
                EntryID [] entryIDsArray =
                    entryIDs.toArray (new EntryID [count]);

                peer.sendMessage (MessageUtils.createEntriesRequestMessage(
                    entryIDsArray));

                requestedEntries += count;

                for (EntryID entryID: entryIDsArray)
                    requestedEntryIDs.put (entryID, peer);

                for (PeerWrapper peerWrapper: peers.values ())
                    peerWrapper.entriesRequested (entryIDsArray);
            }
        }
    }
}
